package org.budo.dubbo.protocol.async.repository.activemq;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Session;

import org.apache.activemq.ActiveMQSession;
import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.budo.support.lang.util.StringUtil;
import org.springframework.beans.factory.DisposableBean;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author limingwei
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractActiveMQAsyncRepository extends AbstractAsyncRepository implements DisposableBean {
    /**
     * 发布/订阅
     */
    public static final String TOPIC = "Topic";

    /**
     * 点对点
     */
    public static final String QUEUE = "Queue";

    private ConnectionFactory connectionFactory;

    private String destinationType;

    /**
     * 消息确认模式，默认为单条消息确认
     */
    private Integer acknowledgeMode = ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE;

    private Map<String, BudoActiveMQSession> interfaceSessionMap = new ConcurrentHashMap<String, BudoActiveMQSession>();

    protected BudoActiveMQSession getSession(String destinationName) {
        BudoActiveMQSession activeMQSession = interfaceSessionMap.get(destinationName);
        if (null != activeMQSession) {
            return activeMQSession;
        }

        synchronized (this) {
            activeMQSession = interfaceSessionMap.get(destinationName);
            if (null != activeMQSession) {
                return activeMQSession;
            }

            activeMQSession = this.initSession(destinationName);
            interfaceSessionMap.put(destinationName, activeMQSession);
            return activeMQSession;
        }
    }

    private BudoActiveMQSession initSession(String destinationName) {
        try {
            Connection connection = this.getConnectionFactory().createConnection(); // 一个接口一个连接
            Session session = connection.createSession(false, this.getAcknowledgeMode()); // 通过 acknoledge方法 签收消息

            Destination destination;

            if (StringUtil.equalsIgnoreCase(this.getDestinationType(), TOPIC)) {
                destination = session.createTopic(destinationName);
            } else {
                destination = session.createQueue(destinationName);
            }

            BudoActiveMQSession budoActiveMQSession = new BudoActiveMQSession(destination, session, connection);
            log.info("#80 initActiveMQSessionOfInterface, destinationName=" + destinationName + ", budoActiveMQSession=" + budoActiveMQSession + ", this=" + this);

            return budoActiveMQSession;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void destroy() throws Exception {
        for (BudoActiveMQSession activeMQSession : this.interfaceSessionMap.values()) {
            log.warn("#93 destroy, this=" + this + ", close activeMQSession=" + activeMQSession);
            activeMQSession.close();
        }

        log.warn("#97 destroy, this=" + this + " done");
    }
}